package ru.ifmo.croak.impl.sharding.team3;

import ru.ifmo.croak.api.MessageListener;
import ru.ifmo.croak.api.MessageQuery;
import ru.ifmo.croak.api.QueryRequest;
import ru.ifmo.croak.api.Message;

import java.util.*;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;

public class CompositeMessageQuery implements MessageQuery, MessageListener {

    final QueryRequest request;
    final MessageQuery[] queries;
    volatile MessageListener listener;
    ExecutorService executor;

    CompositeMessageQuery(ShardingImpl parent, QueryRequest request, MessageQuery[] queries) {
        this.queries = queries;
        this.request = request;
        this.executor = parent.getExecutor();
        
        for(MessageQuery query: queries) {
            query.setMessageListener(this);
        }
    }

    @Override
    public void messageAvailable(ru.ifmo.croak.api.Message message) {
        MessageListener currentListener = listener;
        if (currentListener != null) {
            currentListener.messageAvailable(message);
        }
    }

    @Override
    public QueryRequest getRequest() {
        return request;
    }

    @Override
    public List<Message> getMessages() {
        List<Message> all = new ArrayList<Message>();

        List<Future<List<Message>>> futures = new ArrayList<Future<List<Message>>>();

        for(final MessageQuery query: queries) {
            futures.add(executor.submit(new Callable<List<Message>>() {
                @Override
                public List<Message> call() throws Exception {
                    return query.getMessages();
                }
            }));
        }

        for(Future<List<Message>> fm: futures) {
            List<Message> messages = null;

            try {
                messages = fm.get();
            } catch (InterruptedException e) {
                e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
            } catch (ExecutionException e) {
                e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
            }

            assert messages != null;
            all.addAll(messages);
        }

        if (request.getLimit() > 0) {
            // get `limit` most recent
            Collections.sort(all, new Comparator<Message>() {
                @Override
                public int compare(Message message, Message message1) {
                    long d = message.getTime() - message1.getTime();
                    return d > 0 ? -1 : d < 0 ? 1 : 0;
                }
            });

            all = all.subList(0, Math.min(request.getLimit(), all.size()));
        }

        return all;
    }

    @Override
    public void close() {
        for(MessageQuery query: queries) {
            query.close();
        }

        listener = null;
    }

    @Override
    public void setMessageListener(MessageListener listener) {
        this.listener = listener;
    }
}
